package com.gugee.ins.data.analysis.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.RedisKeyConst;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.dic.InsDicCategoryDao;
import com.gugee.ins.common.dal.service.spider.SpiderSvc;
import com.gugee.ins.common.model.spider.blogger.InsBloggerByIdResponse;
import com.gugee.ins.common.model.spider.blogger.InsBloggerUserFromId;
import com.gugee.ins.data.analysis.config.InsAnalysisConfig;
import com.gugee.ins.data.analysis.service.InsAnalysisAccountCheckUnusualService;
import com.gugee.ins.data.analysis.service.InsAnalysisAccountNormalService;
import com.gugee.ins.data.common.constant.EventTypeEnum;
import com.gugee.ins.data.common.msg.EventMsg;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.ForkJoinPool;

/**
 * InsAnalysisAccountCheckUnusualImpl 服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsAnalysisAccountCheckUnusualImpl implements InsAnalysisAccountCheckUnusualService {

    /**
     * 配置
     */
    @Autowired
    InsAnalysisConfig insAnalysisConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * redis
     */
    @Autowired
    RedisTemplate<String,String> redisTemplate;

    /**
     *
     */
    @Autowired
    InsAnalysisAccountNormalService insAnalysisAccountNormalService;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     *
     */
    @Autowired
    SpiderSvc spiderSvc;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     *
     */
    @Autowired
    InsDicCategoryDao insDicCategoryDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsAnalysisAvatar;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long, EventMsg> kafkaTemplateMonitorAccount;

    /**
     * 线程名
     */
    private String workName = "insActChk404";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insAnalysisConfig.getInsAnalysisAccountCheckUnusualCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 爬取 uid 验证是否异常
     * @param insBloggerMessages
     */
    @Override
    public void handleAnalysisAccountCheckUnusual(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    analysisAccountCheckUnusual(msg);
                } catch (Exception ex) {
                    log.error("insAnalysisAccountCheckUnusual error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("insAnalysisAccountCheckUnusual thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 爬取 uid 验证是否异常
     *
     * @param insBloggerMessage
     */
    public void analysisAccountCheckUnusual(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();
        // redis key
        String key = RedisKeyConst.INS_SPIDER_UNUSUAL + uid;
        // 查询 redis 查看该uid 是否依旧存在，存在则爬取uid接口验证
        String value = redisTemplate.opsForValue().get(key);
        if(!StringUtils.isEmpty(value)){
            // 爬取uid 接口
            try{
                InsBloggerByIdResponse insBloggerByIdResponse = spiderSvc.spiderUid4CheckSuspicious(uid);
                // 判断是接口结果body是否为404
                if(!"-7404".equals(insBloggerByIdResponse.getStatus())){
                    // 判断是不是cookie limit
                    if("-9001".equals(insBloggerByIdResponse.getStatus())){
                        // 重发信息
                        reSendMsg(insBloggerMessage);
                    } else {
                        // 成功，删除redis key
                        insAnalysisAccountNormalService.analysisAccountNormal(uid);
                        // 更新数据
                        if(insBloggerByIdResponse.getUser() != null
                                && insBloggerByIdResponse.getUser().getIsPrivate() != null){
                            updateBlogger(insBloggerByIdResponse.getUser());
                        }
                    }
                } else {
                    // 为null,则redis值加一，判断此时是否超过阈值，是则进行销号处理
                    value = redisTemplate.opsForValue().get(key);
                    if(!StringUtils.isEmpty(value)){
                        checkUnusual(key,uid);
                    }
                }
            }catch (Exception e){
                log.error("insAnalysisAccountCheckUnusual spider uid error,uid:{},msg:{}",uid,e.getMessage());
                // 重发信息
                reSendMsg(insBloggerMessage);
            }
        }
        if ((System.currentTimeMillis() - start) / 1000 > insAnalysisConfig.getInsAnalysisDebugTimeLimit()) {
            log.info("insAnalysisAccountCheckUnusual uid:{} use time:{}s", uid,(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 更新
     * @param uid
     */
    public void updateAccountStatus(long uid){
        insBloggerDao.updateAccountStatus(uid,1);
        // 判断大博主表有没有
        boolean isExist = insBloggerMainDao.findExist(uid);
        if(isExist){
            insBloggerMainDao.updateAccountStatus(uid,1);
        }
    }

    /**
     * 重发
     * @param insBloggerMessage
     */
    public void reSendMsg(InsBloggerMessage insBloggerMessage){
        if(insBloggerMessage.getRetryTimes() > 1){
            insBloggerMessage.setRetryTimes(insBloggerMessage.getRetryTimes() - 1);
            kafkaTemplateInsAnalysisAvatar.send(insAnalysisConfig.getInsBloggerAccountCheckUnusualOnlineTopic(),insBloggerMessage);
        }
    }

    /**
     * 更新播主信息
     * @param insBloggerUserFromId
     */
    public void updateBlogger(InsBloggerUserFromId insBloggerUserFromId){
        try{
            if(!StringUtils.isEmpty(insBloggerUserFromId.getCategory())){
                String overallCategory = insDicCategoryDao.findOverallCategoryByCategory(insBloggerUserFromId.getCategory());
                insBloggerUserFromId.setOverallCategoryName(overallCategory);
            }
            boolean isExist = insBloggerMainDao.findExist(insBloggerUserFromId.getUid());
            if(isExist){
                insBloggerMainDao.updateMBlogger(insBloggerUserFromId);
            }
            insBloggerDao.updateBlogger(insBloggerUserFromId);
        }catch (Exception e){
            log.error("insAnalysisAccountCheckUnusual update blogger error,uid:{},msg:{}",insBloggerUserFromId.getUid(),e.getMessage());
        }
    }


    /**
     * 确认是否销号
     * @param key
     * @param uid
     */
    @Override
    public void checkUnusual(String key,long uid){
        Long nowCount = redisTemplate.opsForValue().increment(key);
        if(nowCount != null && nowCount >= insAnalysisConfig.getInsAnalysisAccountCheckUnusualCount()){
            long now = System.currentTimeMillis();
            // 销号
            redisTemplate.delete(key);
            updateAccountStatus(uid);
            // 发送日志消息给kafka
            sendMonitorMsg(uid,nowCount,now);
        }
    }

    /**
     * 发送日志消息给kafka
     * @param uid
     */
    public void sendMonitorMsg(long uid,long count,long startTime){
        long nowTime = System.currentTimeMillis();
        EventMsg eventMsg = new EventMsg();
        eventMsg.setContent("uid: " + uid + ",调用uid接口返回body为null " + count + " 次,达到阈值("
                + insAnalysisConfig.getInsAnalysisAccountCheckUnusualCount() + "),进行伪删除");

        eventMsg.setOwner(uid + "");
        eventMsg.setEventType(EventTypeEnum.ACCOUNT_PSEUDO_DELETE);
        eventMsg.setPosition("调用uid接口连续返回body为null次数：" + count);
        eventMsg.setSpendMills(nowTime - startTime);
        eventMsg.setId(nowTime);
        kafkaTemplateMonitorAccount.send(insAnalysisConfig.getInsMonitorTopic(),eventMsg);
    }
}
